package wsmsgsubpub

import (
	"dgo/goutils/utils"
	"fmt"
	"gitee.com/ymofen/gobase"
	"io"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"golang.org/x/net/websocket"
)

type WsClient struct {
	sessionid string
	startflag int32
	closeflag int32
	objDNA    uint64

	lkw sync.RWMutex

	closeWg sync.WaitGroup

	ws_req_url string
	origin_url string
	connected  int
	remoteAddr string // 连接的远程地址

	terminated     int
	conn           *websocket.Conn
	dismsg         string
	OnConnectFail  func(sender *WsClient, err error)
	OnSent         func(sender *WsClient, msgtype byte, data []byte)
	OnAfterWrite   func(sender *WsClient, msgtype byte, msg []byte, err error)
	OnRecv         func(sender *WsClient, msgtype byte, data []byte)
	OnDisconnected func(sender *WsClient)
	OnConnected    func(sender *WsClient)
}

func NewWsClient() *WsClient {
	rval := &WsClient{
		objDNA: utils.NewObjectDNA64(),
	}
	return rval
}

func (this *WsClient) RemoteAddr() string {
	return this.remoteAddr
}

func (this *WsClient) doConnectFail(err error) {
	evt := this.OnConnectFail
	if evt != nil {
		evt(this, err)
	}
}

func (this *WsClient) checkConnect() bool {
	if this.conn != nil {
		return true
	}

	wsurl := this.ws_req_url

	if len(wsurl) == 0 {
		this.remoteAddr = ""
		this.doConnectFail(fmt.Errorf("缺少连接参数:requrl"))
		return false
	}

	conn, err := websocket.Dial(wsurl, "", this.origin_url)
	if err != nil {
		this.remoteAddr = ""
		this.doConnectFail(err)
		return false
	}
	this.conn = conn
	this.remoteAddr = conn.RemoteAddr().String()
	evt := this.OnConnected
	if evt != nil {
		evt(this)
	}
	this.closeflag = 0
	this.connected = 1
	return true
}

func (this *WsClient) IsConnected() bool {
	return this.connected == 1
}

func (this *WsClient) RequestDisconnect(msg string) {
	conn := this.conn
	if conn == nil {
		return
	}
	if atomic.CompareAndSwapInt32(&this.closeflag, 0, 1) {
		this.dismsg = fmt.Sprintf("[%s]%s", utils.NowString(), msg)
	}
	conn.Close()
}

func (this *WsClient) doClose() {
	conn := this.conn
	if conn == nil {
		return
	}

	if this.connected == 0 {
		return
	}

	evt := this.OnDisconnected
	if evt != nil {
		evt(this)
	}
	this.connected = 0
	this.conn = nil
}

func (this *WsClient) Stop() {
	this.terminated = 1
	this.RequestDisconnect("停止服务")
}

func (this *WsClient) WaitFor(timeout time.Duration) bool {
	t1 := time.Now()
	for {
		if this.startflag == 0 {
			return true
		}
		if time.Now().Sub(t1) >= timeout {
			return false
		}
		time.Sleep(time.Millisecond * 100)
	}
}

func (this *WsClient) GetWsReqUrl() string {
	return this.ws_req_url
}

func (this *WsClient) ConfigWsReqUrl(ws_req_url, origin_url string) bool {
	isChanged := false
	if this.ws_req_url != ws_req_url {
		this.ws_req_url = ws_req_url
		isChanged = true
	}

	if len(origin_url) == 0 {
		protocol := "http://"
		tmpUrl := ws_req_url
		if strings.HasPrefix(tmpUrl, "ws://") {
			tmpUrl = strings.TrimPrefix(tmpUrl, "ws://")
		} else if strings.HasPrefix(tmpUrl, "wss://") {
			tmpUrl = strings.TrimPrefix(tmpUrl, "wss://")
			protocol = "https://"
		}
		strs := strings.SplitN(tmpUrl, "/", 2)
		origin_url = protocol + strs[0]

		if this.origin_url != origin_url {
			this.origin_url = origin_url
			isChanged = true
		}
	}
	if isChanged {
		this.RequestDisconnect("连接参数改变")
	}

	return isChanged
}

func (this *WsClient) CheckStart() bool {
	if !atomic.CompareAndSwapInt32(&this.startflag, 0, 1) {
		return false
	}
	this.closeWg.Add(0)
	go this.innerDo(func() {
		this.closeWg.Done()
	})
	return true
}

func (this *WsClient) sendWsMessage(conn *websocket.Conn, msgtype byte, msg []byte) error {
	if conn == nil {
		return io.ErrClosedPipe
	}
	this.lkw.Lock()
	defer this.lkw.Unlock()
	writer, err := conn.NewFrameWriter(msgtype)
	if err != nil {
		return err
	}
	_, err = writer.Write(msg)
	if err != nil {
		this.RequestDisconnect(err.Error())
	}
	return err
}

func (this *WsClient) sendPingMessage(conn *websocket.Conn) error {
	if conn == nil {
		return io.ErrClosedPipe
	}
	this.lkw.Lock()
	defer this.lkw.Unlock()
	writer, err := conn.NewFrameWriter(websocket.PingFrame)
	if err != nil {
		return err
	}
	_, err = writer.Write([]byte{})
	return err
}

func (this *WsClient) receiveMessage(conn *websocket.Conn) (msgtype byte, msg []byte, err error) {
again:
	frame, err := conn.NewFrameReader()
	if err != nil {
		return 0, nil, err
	}
	frame, err = conn.HandleFrame(frame)
	if err != nil {
		return 0, nil, err
	}
	if frame == nil {
		goto again
	}
	maxPayloadBytes := conn.MaxPayloadBytes
	if maxPayloadBytes == 0 {
		maxPayloadBytes = websocket.DefaultMaxPayloadBytes
	}
	payloadType := frame.PayloadType()
	data, err := io.ReadAll(frame)
	if err != nil {
		return 0, nil, err
	}
	return payloadType, data, err
}

func (this *WsClient) innerDo(done func()) {
	if gobase.GoFunCatchException {
		defer gobase.DeferCatchPanic()
	}

	defer func() {
		this.startflag = 0
		done()
	}()

	this.terminated = 0
	for {
		if this.terminated == 1 {
			this.RequestDisconnect("服务停止")
			this.doClose()
			break
		}

		if !this.checkConnect() {
			time.Sleep(time.Second * 10)
			continue
		}

		msgtype, msg, err := this.receiveMessage(this.conn)
		if err != nil {
			this.RequestDisconnect(fmt.Sprintf("接收出现异常:%s", err.Error()))
			this.doClose()
			continue
		}

		if this.OnRecv != nil {
			this.OnRecv(this, msgtype, msg)
		}
	}
}

func (this *WsClient) SendMsg(msgtype byte, msg []byte) error {
	return this.sendWsMessage(this.conn, msgtype, msg)
}

func (this *WsClient) Ping() error {
	return this.sendPingMessage(this.conn)
}
